package membership;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.Collections;
import java.util.List;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

import remoteprocedurecall.RPCClient;
import remoteprocedurecall.RPCDestination;

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.services.simpledb.AmazonSimpleDB;
import com.amazonaws.services.simpledb.AmazonSimpleDBClient;
import com.amazonaws.services.simpledb.model.Attribute;
import com.amazonaws.services.simpledb.model.BatchPutAttributesRequest;
import com.amazonaws.services.simpledb.model.CreateDomainRequest;
import com.amazonaws.services.simpledb.model.DeleteAttributesRequest;
import com.amazonaws.services.simpledb.model.DeleteDomainRequest;
import com.amazonaws.services.simpledb.model.Item;
import com.amazonaws.services.simpledb.model.PutAttributesRequest;
import com.amazonaws.services.simpledb.model.ReplaceableAttribute;
import com.amazonaws.services.simpledb.model.ReplaceableItem;
import com.amazonaws.services.simpledb.model.SelectRequest;
import com.amazonaws.services.simpledb.model.SelectResult;

public class SimpleDBMembership {

	// local variables
	public String myIpAddress;
	public Integer myPort;
	Random rand = new Random();

	// cached server list
	public static List<RPCDestination> servers = new ArrayList<RPCDestination>();

	// simpleDB call variables
	private final static String myDomain = "ssmMembership";
	private final static String myMembershipItem = "servers";
	private final static String myMembershipItemStatusRunning = "running";

	// instance
	private static SimpleDBMembership sdm;

	public static SimpleDBMembership getInstance() {
		if (sdm == null) {
			sdm = new SimpleDBMembership();
		}
		return sdm;
	}

	public static SimpleDBMembership getInstance(String ipAddress, Integer port) {
		if (sdm == null) {
			sdm = new SimpleDBMembership(ipAddress, port);
		}
		if (sdm.myIpAddress == null) {
			sdm.myIpAddress = ipAddress;
			sdm.myPort = port;
			Thread thread = new Thread() {
				public void run() {
					sdm.membershipLoop();
				}
			};
			System.out.println("Started membership loop at " + ipAddress + ":"
					+ port);
			thread.start();
		}
		return sdm;
	}

	/**
	 * Constructor
	 * 
	 * @param ipAddress
	 *            - ip of the RPCServer
	 * @param port
	 *            - port of the RPCServer
	 */
	private SimpleDBMembership(String ipAddress, Integer port) {
		this.myIpAddress = ipAddress;
		this.myPort = port;
	}

	private SimpleDBMembership() {

	}

	// Membership thread
	private void membershipLoop() {
		AmazonSimpleDB client = null;
		while (client == null) {
			// System.out.println("Trying to get client");
			// Set up SDB client
			AWSCredentials awsCredentials = new BasicAWSCredentials(
					"AKIAJAMYO5EPQY5VQ4DA",
					"fH96pmQ7LhswDb9ORvZK7JGj+b3gplhAbyBa8R0M");
			client = new AmazonSimpleDBClient(awsCredentials);
			// initialize domain if it has not been set up yet

			try {
				//ssmDeleteDomain(client);// TODO: remove in final version
				ssmCreateDomain(client);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

		// Membership loop
		while (true) {
			try {

				// retrieve list of servers, synchronize to avoid local data
				// contention
				synchronized (servers) {
					servers.clear();
					servers = ssmGetServers(client);
					// make sure I am in this list
					boolean inList = false;
					for (RPCDestination rd : servers) {
						// System.out.println("checking for server in db " +
						// rd.destAddress + " : " + rd.destPort);

						// System.out.println("my information is " + myIpAddress
						// + " : " +myPort);
						if (rd.destAddress.startsWith(myIpAddress)
								&& ("" + rd.destPort).startsWith(("" + myPort))) {
							System.out.println("Set to true");
							inList = true;
						}
					}
					// write myself to SDB and servers if i'm NOT in the list
					// System.out.println("inlist is " + inList);
					if (!inList) {
						// System.out.println("Adding to list...");

						servers.add(new RPCDestination(myIpAddress, myPort));
						ssmAddMembership(client, myIpAddress,
								String.valueOf(myPort));
					}
				}

				// Check if random server is up and running
				int randServerNum = rand.nextInt(servers.size());
				RPCDestination dest = servers.get(randServerNum);
				DatagramSocket rpcSocket = null;
				// if(!(dest.destAddress.equals(myIpAddress) && dest.destPort ==
				// myPort)){
				// System.out.println("pinging " + dest.destAddress + " : " +
				// dest.destPort);
				try {
					rpcSocket = new DatagramSocket();
					String ping = "0" + RPCClient.RPC_SEP + RPCClient.OP_PING;
					// send ping
					DatagramPacket sendPkt = new DatagramPacket(
							ping.getBytes(), ping.getBytes().length,
							InetAddress.getByName(dest.destAddress),
							dest.destPort);
					rpcSocket.send(sendPkt);
				} catch (Exception e) {
				}

				// Recieve response - if no response, remove server from servers
				// and SDB
				byte[] inBuf = new byte[1024];
				try {
					DatagramPacket recvPkt = new DatagramPacket(inBuf,
							inBuf.length);
					recvPkt.setLength(inBuf.length);
					rpcSocket.setSoTimeout(5000); // Set timeout to 5 seconds
					// recieve packet
					rpcSocket.receive(recvPkt);
					String receivedString = new String(inBuf);
					// System.out.println("pong receive:"+receivedString);

					if (!receivedString.startsWith("0" + RPCClient.RPC_SEP
							+ RPCClient.OP_PONG)) {
						// If it doesn't respond with PONG then delete it from
						// the server list
						// System.out.println("deleting " +dest.destAddress +
						// " : " + dest.destPort );

						ssmDeleteServer(client, dest.destAddress,
								String.valueOf(dest.destPort));
						for (RPCDestination sd : servers) {
							if (sd.destAddress.equals(dest.destAddress)
									&& sd.destPort == dest.destPort) {
								servers.remove(sd);
								break;
							}
						}
					}
				} catch (Exception e) {
					// If error, assume the randomly chosen server has failed
					ssmDeleteServer(client, dest.destAddress,
							String.valueOf(dest.destPort));
					for (RPCDestination sd : servers) {
						if (sd.destAddress.equals(dest.destAddress)
								&& sd.destPort == dest.destPort) {
							servers.remove(sd);
							break;
						}
					}
				}

				if (rpcSocket != null)
					rpcSocket.close();
			} catch (Exception e) {
				// could not connect to DB!
			}
			
			// sleep 5-10 seconds
			try {
				Thread.sleep(rand.nextInt(5000) + 5000);
			} catch (InterruptedException e1) {
				System.err
						.println("Could not sleep for in hear beat thread");
				e1.printStackTrace();
			}
		}

	}

	/**
	 * ssmSetMyMembership - adds the given server to simpleDB
	 * 
	 * @param sdb
	 *            - simpleDB instance
	 * @param ip
	 *            - ip of server to add
	 * @param port
	 *            - port of server to add
	 * @throws Exception
	 */
	private void ssmAddMembership(AmazonSimpleDB sdb, String ip, String port)
			throws Exception {
		// Put data into a domain
		List<ReplaceableItem> sampleData = new ArrayList<ReplaceableItem>();
		sampleData.add(new ReplaceableItem(myMembershipItem)
				.withAttributes(new ReplaceableAttribute(ip + ":" + port,
						myMembershipItemStatusRunning, true)));
		sdb.batchPutAttributes(new BatchPutAttributesRequest(myDomain,
				sampleData));
	}

	/**
	 * ssmDeleteServer - deletes a particular server from myDomain
	 * 
	 * @param sdb
	 *            - simpleDB instance
	 * @param ip
	 *            - ip of server to remove
	 * @param port
	 *            - port of server to remove
	 * @throws Exception
	 */
	private void ssmDeleteServer(AmazonSimpleDB sdb, String ip, String port)
			throws Exception {
		// Put data into a domain
		sdb.deleteAttributes(new DeleteAttributesRequest(myDomain,
				myMembershipItem).withAttributes(new Attribute(ip + ":" + port,
				myMembershipItemStatusRunning)));
	}

	/**
	 * ssmDeleteDomain - removes myDomain from simpleDB if it exists
	 * 
	 * @param sdb
	 *            - simpleDB instance
	 * @throws Exception
	 */
	private void ssmDeleteDomain(AmazonSimpleDB sdb) throws Exception {
		// Put data into a domain
		sdb.deleteDomain(new DeleteDomainRequest(myDomain));
	}

	/**
	 * ssmCreateDomain - creates myDomain if it doesn't exist
	 * 
	 * @param sdb
	 *            - simpleDB instance
	 * @throws Exception
	 */
	private void ssmCreateDomain(AmazonSimpleDB sdb) throws Exception {
		// Put data into a domain
		sdb.createDomain(new CreateDomainRequest(myDomain));
	}

	/**
	 * ssmGetServers - gets all servers in the sdb
	 * 
	 * @param sdb
	 *            - simpleDB instance
	 * @return - list of <ip,port> pairs
	 * @throws Exception
	 */
	private List<RPCDestination> ssmGetServers(AmazonSimpleDB sdb)
			throws Exception {
		// consistent read? yes
		boolean consistentRead = true;
		// Get all servers
		List<RPCDestination> servers = new ArrayList<RPCDestination>();
		String selectExpression = "select * from " + myDomain;
		SelectRequest selectRequest = new SelectRequest(selectExpression,
				consistentRead);
		selectRequest.setConsistentRead(true);
		for (Item item : sdb.select(selectRequest).getItems()) {
			for (Attribute attribute : item.getAttributes()) {
				servers.add(new RPCDestination(
						attribute.getName().split(":")[0], Integer
								.parseInt(attribute.getName().split(":")[1])));
			}
		}
		return servers;
	}

	/**
	 * getNServers - gets N random servers (or as many as possible up to N)
	 * 
	 * @param N
	 *            - number of server locations desired
	 * @return - list of servers
	 */
	public List<RPCDestination> getNServers(Integer N) {
		List<RPCDestination> serverList = new ArrayList<RPCDestination>();
		// shuffling and returning the first N elements of servers is
		// sufficiently random
		Collections.shuffle(servers);
		for (int i = 0; i < N && i < servers.size(); ++i) {
			serverList.add(new RPCDestination(servers.get(i).destAddress,
					servers.get(i).destPort));
		}
		return serverList;
	}
}
